package com.allscore.storm.demo.kafka;


import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import storm.kafka.*;

import java.util.Arrays;

public class MyKafkaTopology {
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {

        String zks = "127.0.0.1:2181";
        String topic = "test0001";
        String id = "wordtest";
        String[] zkServerArray = new String[]{"127.0.0.1"};

        BrokerHosts brokerHosts = new ZkHosts(zks);
        SpoutConfig spoutConf = new SpoutConfig(brokerHosts, topic, "", id);
        spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
        spoutConf.forceFromStart = false;
        spoutConf.useStartOffsetTimeIfOffsetOutOfRange = true;
        spoutConf.zkServers = Arrays.asList(zkServerArray);
        spoutConf.zkPort = 2181;

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafka-reader", new KafkaSpout(spoutConf), 2); // Kafka我们创建了一个2分区的Topic，这里并行度设置为2
        builder.setBolt("print-bolt", new PrintBolt(), 2).shuffleGrouping("kafka-reader");

        Config conf = new Config();
        String name = MyKafkaTopology.class.getSimpleName();
        if (args != null && args.length > 0) {
            // Nimbus host name passed from command line
            conf.put(Config.NIMBUS_HOST, args[0]);
            conf.setNumWorkers(3);
            StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
        } else {
            conf.setMaxTaskParallelism(3);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology(name, conf, builder.createTopology());
        }
    }

}
